package io.openmessaging.storage;

import io.openmessaging.Conf;
import io.openmessaging.Message;
import io.openmessaging.Topic;
import io.openmessaging.io.FileChannelWrapper;
import io.openmessaging.map.TopicQueueOffsetMap;
import io.openmessaging.utils.Assert;
import io.openmessaging.utils.Log;
import io.openmessaging.utils.Statistic;

import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author cuijd    2021/9/26 9:11
 */
public class SSDStorage2 implements Storage {
    private StorageFile[] fileChannels = new StorageFile[10];
    AtomicInteger activeFile = new AtomicInteger();
    private final int writeCount = 10;//每 write writeCount 次执行一次 force

    static class StorageFile {
        FileChannelWrapper channel;
        final Message[] waitNotify = new Message[40];
        AtomicInteger writeStartCount = new AtomicInteger();//上次 force 到现在开始 write 的消息数量
        AtomicInteger writeEndCount = new AtomicInteger();//上次 force 到现在已经 write 完毕的消息数量
    }


    private TopicQueueOffsetMap<FileIndex> indices = new TopicQueueOffsetMap<>();

    public SSDStorage2() {
        for (byte i = 0; i < fileChannels.length; i++) {
            fileChannels[i] = new StorageFile();
            fileChannels[i].channel = new FileChannelWrapper(Conf.ssdPath + i);

            if (fileChannels[i].channel.size() == 0)
                continue;
            //根据已有数据恢复索引
            long readPos = 0;
            ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
            while (readPos < fileChannels[i].channel.size()) {
                head.clear();
                readPos += fileChannels[i].channel.readAndFlip(head, readPos);
                FileIndex index = indices.set(head.get(), head.getShort(), head.getInt(), new FileIndex(i, readPos - 9, head.getShort()));
                readPos += index.length;
            }
        }
    }

    @Override
    public void save(Message m) {
        save(m, activeFile.get());
    }

    public void save(Message m, int fileIdSeq) {
        int fileId = fileIdSeq % fileChannels.length;
        StorageFile file = fileChannels[fileId];
        int startCount = file.writeStartCount.addAndGet(1);

        if (startCount > writeCount) {
            file.writeStartCount.addAndGet(-1);
            int nextFileIdSeq = fileIdSeq + 1;

            boolean success = activeFile.compareAndSet(fileIdSeq, nextFileIdSeq);
//            if (success) {
//                Log.i("切换 activeFile: ", fileId, " -> ", fileId + 1);
//                Assert.isTrue(nextStartCount % writeCount == 0
//                        , "切换 activeFile 后 startCount = " + nextStartCount + " fileId = " + nextFileId);
//            }

//                Log.i("resave waitSize = ", waitSize, " activeFile = ", fileId);
            save(m, nextFileIdSeq);
            return;
        }

        ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
        head.clear();
        head.put(m.topic.id).putShort((short) m.queueId).putInt((int) m.offset).putShort((short) m.data.remaining()).flip();
        //等待唤醒
//        int waitIndex = startCount % writeCount;
//        Assert.isTrue(file.waitNotify[waitIndex] == null, "waitNotify 不干净 fileId = " + fileId + " waitIndex = " + waitIndex);
//        file.waitNotify[waitIndex] = m;
        //写入ssd
        int dataSize = m.data.remaining();
        long position = file.channel.write(head, m.data);
        //消息索引
        FileIndex index = new FileIndex((byte) fileId, position, (short) dataSize);
        indices.set(m.topic.id, m.queueId, m.offset, index);
        int writeEndCount = file.writeEndCount.getAndAdd(1) % writeCount;
        if (writeEndCount == writeCount - 1) {//由最后一个写完的线程进行force
            file.channel.force();
            for (int i = 0; i <= 9; i++) {
//                Log.i("唤醒 ", file.waitNotify[i], " fileId = ", fileId, " waitIndex = ", i);
                synchronized (file.waitNotify) {
                    file.waitNotify.notifyAll();
//                    file.waitNotify[i] = null;
                }
            }
            int before = file.writeStartCount.get();
            while (!file.writeStartCount.compareAndSet(writeCount, 0)) {
                Log.i("重置 waitNotifySize 失败，重试。 fileId = ", fileId, " before = ", before, " resetSuccess = ");
//                try {
//                    Thread.sleep(1);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
            }
//            boolean resetSuccess = file.writeStartCount.compareAndSet(writeCount, 0);
//            Log.i("重置 writeStartCount: fileId = ", fileId, " before = ", before, " resetSuccess = ", resetSuccess);
//            Assert.isTrue(resetSuccess, "重置 waitNotifySize 失败 fileId = ", fileId, " before = ", before, " resetSuccess = ", resetSuccess);
        } else {
            synchronized (file.waitNotify) {

//                    Log.i("等待 ", m, " fileId = ", fileId, " waitIndex = ", waitIndex);
                try {
                    long start = System.currentTimeMillis();
                    file.waitNotify.wait(1000);
                    //最后几条消息如果不够 writeCount 条不会被唤醒，超时自动醒
                    if (System.currentTimeMillis() - start > 900) {
                        file.channel.force();
                        Log.i("等待超时 ", m, " fileId = ", fileId);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    System.exit(2);
                }
            }

        }
    }

    @Override
    public ByteBuffer get(Topic topic, int queueId, long offset) {
        return get(topic, queueId, offset, null);
    }

    /**
     * 先判断有没有落盘，如果没有，返回null，否则先从缓存中取，如果缓存取不到，再从 ssd 读
     */
    public ByteBuffer get(Topic topic, int queueId, long offset, Storage cache) {
        FileIndex index = this.indices.get(topic.id, queueId, offset);
        if (index == null)
            return null;

        ByteBuffer readBuffer = null;
        if (cache != null)
            readBuffer = cache.get(topic, queueId, offset);

        if (readBuffer == null) {
            readBuffer = ByteBuffer.allocate(index.length);
            fileChannels[index.fileId].channel.readAndFlip(readBuffer, index.position + 9);//跳过header

            Statistic.SSD.readCount.add(1);
            Statistic.SSD.readSize.add(readBuffer.remaining());
        }
        return readBuffer;
    }

}
